package com.erainm.custominterceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;

import java.util.Map;

/**
 * Created by erainm on 2021/9/10 8:45 上午.
 *
 * @Description TODO:自定义拦截器1
 **/
public class InterceptorThree implements ProducerInterceptor<Integer,String> {

    @Override
    public ProducerRecord<Integer, String> onSend(ProducerRecord<Integer, String> record) {
        System.out.println("拦截器3 -- go");
        // 要发送的消息内容
        String topic = record.topic();
        Integer partition = record.partition();
        Integer key = record.key();
        String value = record.value();
        Long timestamp = record.timestamp();
        Headers headers = record.headers();
        // 消息发送时，经过拦截器，调用该方法
        // 拦截下来后根据原有消息创建新的消息，这里没有对原消息做任何处理
        ProducerRecord<Integer, String> newRecord = new ProducerRecord<Integer, String>(topic,partition,timestamp,key,value,headers);
        return newRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("拦截器3 -- back");
        // 消息确认或异常的时候，调用该方法，该方法中不应该实现较重的任务
        // 会影像kafka生产者性能
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
        Object classContent = configs.get("classContent");
        System.out.println(classContent);
    }
}
